package com.atguigu.edu.realtime.app.demo;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.util.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

import java.time.Duration;

public class FlinkApiJoinApp {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        String leftTopic="demo_left_topic";
        String rightTopic="demo_right_topic";
        String groupId="demo_join_app";
        DataStreamSource<String> leftStream = env.addSource(MyKafkaUtil.getKafkaConsumer(leftTopic, groupId));
        DataStreamSource<String> rightStream = env.addSource(MyKafkaUtil.getKafkaConsumer(rightTopic, groupId));

        //第一步：将流数据转化成json     开窗会涉及到时间，一般选用事件时间，不要用外部的时间
        SingleOutputStreamOperator<JSONObject> leftObjStream = leftStream.map(JSON::parseObject);
        SingleOutputStreamOperator<JSONObject> rightObjStream = rightStream.map(JSON::parseObject);
        //加时间语义，分配水位线  watermark
        SingleOutputStreamOperator<JSONObject> leftJsonWmStream = leftObjStream.assignTimestampsAndWatermarks(WatermarkStrategy.<JSONObject>forMonotonousTimestamps()
                .withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
                    @Override
                    public long extractTimestamp(JSONObject jsonObject, long recordTimestamp) {
                        return jsonObject.getLong("ts");//提出一个时间来
                    }
                }));
        //顺序得到数据进行join
        //SingleOutputStreamOperator<JSONObject> rightJsonWmStream = rightObjStream.assignTimestampsAndWatermarks(WatermarkStrategy.<JSONObject>forMonotonousTimestamps()

        //为解决乱序问题           ---->   增加两秒的乱序容忍时间
        SingleOutputStreamOperator<JSONObject> rightJsonWmStream = rightObjStream.assignTimestampsAndWatermarks(WatermarkStrategy.<JSONObject>forBoundedOutOfOrderness(Duration.ofMillis(2000))
                .withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
                    @Override
                    public long extractTimestamp(JSONObject jsonObject, long recordTimestamp) {
                        return jsonObject.getLong("ts");//提出一个时间来
                    }
                }));

/*      //*****************window join(开窗时候用到的join,window是手段，join是目的    把数据框在一起，在这个窗口之内做数据比对) ****************************************************

        //订单  {"id":"101","user_id":"u_101","status":"1001", "amount":200,"ts":1000}
        // 订单明细 {"id":"990","order_id":"101","sku_id":"s_77",  "ts":1000}   订单明细的order_id与订单的id相关联
        //inner join
        DataStream<JSONObject> joinedStream = leftJsonWmStream.join(rightJsonWmStream)
                .where(leftJson -> leftJson.getString("id"))
                .equalTo(rightJson -> rightJson.getString("order_id"))
                .window(TumblingEventTimeWindows.of(Time.milliseconds(100000)))//窗口类型选择:滚动 or 滑动，一般选择滚动  此处订单和订单明细时间时间相同，不会出现在不同窗口，所以可以选滚动窗口
                .apply(new JoinFunction<JSONObject, JSONObject, JSONObject>() {//将两个合在一块
                    @Override
                    public JSONObject join(JSONObject leftJson, JSONObject rightJson) throws Exception {
                        JSONObject joinedJson = new JSONObject();
                        joinedJson.put("detail_id", rightJson.getString("id"));
                        joinedJson.put("order_id", leftJson.getString("id"));
                        joinedJson.put("user_id", leftJson.getString("user_id"));
                        joinedJson.put("status", leftJson.getString("status"));
                        joinedJson.put("sku_id", rightJson.getString("sku_id"));
                        joinedJson.put("ts", rightJson.getLong("ts"));
                        return joinedJson;
                    }
                });*/

/*
       //**************************left join*************************************

        //  订单明细
        //   {"id":"991","order_id":"101","amount":200,"ts":1000}
        //   {"id":"992","order_id":"102","amount":400,"ts":1000}
        //   {"id":"993","order_id":"103","amount":100,"ts":11000}
        //   购物券使用
        //   {"id":"7701","order_detail_id":"991","coupon_id":"3301" ,"ts":1000}
        //   {"id":"7702","order_detail_id":"993","coupon_id":"3302" ,"ts":12000}
        DataStream<JSONObject> leftJoinWindowStream = leftJsonWmStream.coGroup(rightJsonWmStream)
                .where(leftJson -> leftJson.getString("id"))
                .equalTo(rightJson -> rightJson.getString("order_detail_id"))
                .window(TumblingEventTimeWindows.of(Time.milliseconds(100000)))//窗口类型选择:滚动 or 滑动，一般选择滚动  此处订单和订单明细时间时间相同，不会出现在不同窗口，所以可以选滚动窗口
                .apply(new CoGroupFunction<JSONObject, JSONObject, JSONObject>() {
                    @Override  ///迭代器  +   迭代器   +   输出
                    public void coGroup(Iterable<JSONObject> leftItr, Iterable<JSONObject> rightItr, Collector<JSONObject> out) throws Exception {
                        //因为不知道左右两边的数量不确定一样多，所以先进行循环迭代
                        for (JSONObject leftJson : leftItr) {
                            boolean ifExistsRight = false;//如果没有也要进行一次输出 --> 判断是否存在右边 ， 默认是不存在
                            for (JSONObject rightJson : leftItr) {//如果右边有数据 --> 两边数据进行合并之后进行输出
                                ifExistsRight = true;//确实有右边的数据存在
                                leftJson.put("coupon_id", rightJson);//右边可以join上数据，选择右边的数据
                                out.collect(leftJson);//关联上之后就可以进行输出操作
                            }
                            if (!ifExistsRight) {//如果右边没有数据 --> 左边默默的输出
                                out.collect(leftJson);//左边为驱动，右边往左边添加，就算右边没有数据左边一样输出
                            }
                        }
                    }
                });
        leftJoinWindowStream.print(" Window left join ::: ");
*/

        //**************************intervaljoin*************************************

        //  订单明细
        //   {"id":"101","user_id":"u_101","status":"1001", "amount":200,"ts":1000}
        //   {"id":"102","user_id":"u_102","status":"1001", "amount":400,"ts":11000}
        //   {"id":"103","user_id":"u_103","status":"1001", "amount":600,"ts":15000}
        //   支付
        //   {"id":"55","order_id":"101","type":"wx" ,"ts":4000}
        //   {"id":"56","order_id":"102","type":"ali" ,"ts":12000}
        KeyedStream<JSONObject, String> leftKeyedStream = leftJsonWmStream.keyBy(leftjson -> leftjson.getString("id"));
        KeyedStream<JSONObject, String> rightKeyedStream = rightJsonWmStream.keyBy(rightjson -> rightjson.getString("order_id"));
        SingleOutputStreamOperator<JSONObject> intervalJoinedStream = leftKeyedStream.intervalJoin(rightKeyedStream)//处理
                .between(Time.milliseconds(-50), Time.seconds(60 * 15 + 5))//支付的实现时间限制
                .process(new ProcessJoinFunction<JSONObject, JSONObject, JSONObject>() {
                    @Override
                    public void processElement(JSONObject left, JSONObject right, Context ctx, Collector<JSONObject> out) throws Exception {
                        left.put("type", right.getString("type"));
                        left.put("pay_ts", right.getLong("ts"));
                        out.collect(left);
                    }
                });

        intervalJoinedStream.print("interval join  :::");
        env.execute();
    }
}
















